package com.leo.hadoop.map;

import com.leo.hadoop.map.*;
import com.leo.util.*;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.conf.Configuration;

import java.util.regex.Pattern;
import java.util.List;
import java.util.ArrayList;

/**
 * Describe class SelectMapper here.
 *
 *
 * Created: Thu Apr 18 22:37:48 2013
 *
 * @author <a href="mailto:leo@leo-pc">Leo</a>
 * @version 1.0
 */
public class SelectMapper extends MyMapper {

    private List<Integer> columns;
    private Integer keyIndex = null;
    /**
     * Creates a new <code>SelectMapper</code> instance.
     *
     */
    public SelectMapper() {

    }

    @Override
    public void process(Text key, Text value, Context context) throws IOException, InterruptedException {
	String result = "";
	for (int index : columns) 
	    result += this.schema.getString(value.toString(), index);

	if (keyIndex != null)
	    key = new Text(this.schema.getString(value.toString(), keyIndex));
	context.write(key, new Text(result));
    }

    @Override
    public Schema getOutputSchema() {
	return this.schema.subSchema(columns);
    }

    @Override
    protected void init(Configuration conf) {
	String input = conf.get("select");
	int keyIndex = conf.getInt("key", 0);
	String regex = "\\d+(,\\d)*";
	
	if (keyIndex != 0 && !this.schema.test(keyIndex))
	    throw new MapperInitException("Key varible illegle.");
	if (input == null || !Pattern.matches(regex, input))
	    throw new MapperInitException("select varible illegle.");
	    
	this.keyIndex = keyIndex;
	columns = new ArrayList<Integer>();
	String[] list = input.split(",");
	for (String column : list) {
	    int a = Integer.parseInt(column);;
	    if (this.schema.test(a))
		columns.add(a);
	    else
		throw new MapperInitException();
	}
    }

}
